#include "I_EventProcessor.h"
#include <string.h>
#include <stdio.h>
#include <zlib/lib/Ink_defs.h>

const int ET_CALL =  0;
EventProcessor eventProcessor;
volatile off_t thread_data_used[MAX_THREAD_GROUP] = {0};

EventProcessor::EventProcessor() : n_thread_group(0), n_ethreads(0), n_dthreads(0)
{
	memset(all_ethreads, 0, sizeof(all_ethreads));
	memset(event_threads, 0, sizeof(event_threads));
	memset(all_dthreads, 0, sizeof(all_dthreads));
	memset(n_threads_for_type, 0, sizeof(n_threads_for_type));
	memset(next_thread_for_type, 0, sizeof(next_thread_for_type));

	ink_mutex_init(&d_spaw_thread_mutex);
}

EventProcessor::~EventProcessor()
{

}

/*all processor start and spawn function is  non - reentrant, this function should be call in main*/
bool
EventProcessor::start(int number_of_threads, int detached, size_t stacksize)
{
	if(number_of_threads > MAX_TYPE_THREADS)
		assert(!"eventprocessor start thread overstep MAX_TYPE_THREADS");

	char thread_name[MAX_THREAD_NAME_LEN];

	for(int i = 0; i < number_of_threads; i++)
	{
		EThread *t = new EThread(i);
		all_ethreads[n_ethreads++] = t;
		event_threads[ET_CALL][i] = t;
	}

	n_threads_for_type[ET_CALL] = number_of_threads;
	n_thread_group++;

	for(int i = 0; i < number_of_threads; i++)
	{
		snprintf(thread_name, MAX_THREAD_NAME_LEN - 1, "[ET_CALL %d]", i);
		event_threads[ET_CALL][i]->start(thread_name, NULL, NULL, detached, stacksize);
	}

	return true;
}


EventType
EventProcessor::spawn_event_threads(int n_threads, const char *et_name, int detached, size_t stacksize)
{
	if(n_threads > MAX_TYPE_THREADS)
		assert(!"eventprocessor spawn event thread overstep MAX_TYPE_THREADS");
	if(n_thread_group > MAX_THREAD_GROUP)
		assert(!"eventprocessor spawn event thread more thran MAX_THREAD_GROUP");

	char thread_name[MAX_THREAD_NAME_LEN];

	EventType new_thread_type = n_thread_group;

	for(int i = 0; i < n_threads; i++)
	{
		EThread *t = new EThread(i);
		all_ethreads[n_ethreads++] = t;
		event_threads[new_thread_type][i] = t;
	}

	n_threads_for_type[new_thread_type] = n_threads;
	n_thread_group++;

	for(int i = 0; i < n_threads; i++)
	{
		snprintf(thread_name, MAX_THREAD_NAME_LEN - 1, "[%s %d]", et_name, i);
		event_threads[new_thread_type][i]->start(thread_name, NULL, NULL, detached, stacksize);
	}

	return new_thread_type;
}

//no detached
//A D Type Thread may be create in event thread
//So it's probably a multithreading operation here
Event *
EventProcessor::spawn_thread(Continuation *cont, const char *thr_name)
{
	if(n_dthreads > MAX_DETACHED_THREADS)
		assert(!"the independent threads outside the upper");
	
	Event *e = new Event;
	e->init(cont, 0, 0);
	e->ethread = new EThread(e);
	e->mutex = cont->mutex = e->ethread->mutex_ptr;

	ink_mutex_lock(&d_spaw_thread_mutex);
	all_dthreads[n_dthreads++] = e->ethread;
	ink_mutex_release(&d_spaw_thread_mutex);

	e->ethread->start(thr_name, NULL, NULL);

	return NULL;
}

void
EventProcessor::shutdown()
{
//TODO:
//probe thread alive
//If more than a period of time ,And thread	has alive
//we should be froce kill this.
}

EThread *
EventProcessor::assign_thread(EventType et)
{
	int next;

	if(et > MAX_THREAD_GROUP)
		assert(!"overstep MAX_THREAD_GROUP");

	if(n_threads_for_type[et] > 1)
	{
		next = next_thread_for_type[et];
		next_thread_for_type[et]++;
		next_thread_for_type[et] %= n_threads_for_type[et];
	}else{
		next = 0;
	}

	return event_threads[et][next];
}

Event *
EventProcessor::schedule_imm(Continuation *c, EventType et, int callback_event)
{
	Event *e = new Event;
	e->callback_event = callback_event;
	return schedule(e->init(c, 0, 0), et);
}

Event *
EventProcessor::schedule_imm_signal(Continuation *c, EventType et, int callback_event)
{
	Event *e = new Event;
	e->callback_event = callback_event;
	return schedule(e->init(c, 0, 0), et, true);
}

Event *
EventProcessor::schedule_at(Continuation *c, ink_hrtime timeout_at, EventType et, int callback_event)
{
	Event *e = new Event;
	e->callback_event = callback_event;
	return schedule(e->init(c, timeout_at, 0), et);
}

Event *
EventProcessor::schedule_every(Continuation *c, ink_hrtime period, EventType et, int callback_event)
{
	Event *e = new Event;
	e->callback_event = callback_event;
	//timeout_at = period
	return schedule(e->init(c, period, period), et);
}

Event *
EventProcessor::schedule(Event *e, EventType etype, bool fast_signal)
{
	e->ethread = this->assign_thread(etype);

	if(!e->continuation->mutex.m_ptr)
		e->mutex = e->continuation->mutex = e->ethread->mutex_ptr;
	else
		e->mutex = e->continuation->mutex;

	e->ethread->notifyQueue.enqueue(e, fast_signal);
	return e;
}

off_t
EventProcessor::allocate_thread_spec(int size, EventType etype)
{
	off_t start;
	//16 byte alignment
	size = INK_ALIGN(size, 16);
	do{
		start = thread_data_used[etype];
		if(start + size > THREAD_PRIVATE_DATA) {
			assert(!"thread private spec exhausted!");
			return -1;
		}
	}while(!ink_atomic_cas(&thread_data_used[etype], start, start+size));

	return start;
}
